/**
 *  by Michael Liljenstam (ML). revisions ato.
 */

package SSF.Net;

import SSF.OS.*;
import SSF.OS.NetFlow.BytesUtil;
import SSF.Util.Streams.*;
import com.renesys.raceway.DML.*;
import com.renesys.raceway.SSF.*;
import java.text.*;
import java.util.LinkedList;

/** A prototype queue monitor built atop the SSF.Util.Streams
 * infrastructure, provided as a demonstration.
 * Monitors the IP packet arrivals and drops in a droptailQueue
 * on the NIC on which this droptailQueueMonitor is configured.
 * Computes the average queue length as the time integral of queue length
 * in consecutive time intervals of length <TT>probe_interval</TT>
 * and writes to the named StreamInterface.
 *
 * <P>Uses record_type(s): SSF.Net.QueueRecord_1, SSF.Net.QueueProbeIntRecord.
 *
 * @see SSF.Net.droptailQueue
 * @see SSF.Net.droptailRecordPlayer_1
 * @see SSF.Util.Streams
 */
public class droptailQueueMonitor_1 implements PacketQueueMonitor {
  private java.text.DecimalFormat fm = new DecimalFormat();
  private droptailQueue queue;
  private NIC nic;
  private long tickinterval;
  private boolean DEBUG=false;
  private Host localHost;
  private boolean suppressZeros = false;
  private double invfreq;         // inverse Net.frequency: sec/tick
  private double intervalLength;  // in seconds
  private double qlen = 0;        // instantaneous queue length in bits
  private double bitrate;

  private ProbeSession P;
  private StreamInterface S;
  private int hostcode;
  private int typeCode;

  private int pkts = 0;             // in current interval
  private int drops = 0;            // in current interval
  private long prevFlakyDrops = 0;  // flaky drops in previous interval

  /*
   * In order to compute time-averaging statistics for the queue, rather than 
   * just pick samples periodically, we need per-packet information. So for 
   * each arriving packet a packetQueueRecord is kept. This allows the monitor 
   * to calculate the time integral of the queue length. Not optimized.
   */
  class PacketQueueRecord {
    public double  arrivalTime; // in seconds
    public int   size;          // in bytes
    //public boolean dropped;

    public PacketQueueRecord(double _arrivalTime, int _size) { //, boolean _dropped) {
      arrivalTime = _arrivalTime;
      size = _size;
      //dropped = _dropped;
    }
  }

  /* List of packetQueueRecords kept until the next sample interval, when
   *  the relevant statistics are computed and written to a Recorder stream.
   */
  private LinkedList packetQueueLog;
  private long qResidual;

  /** configure the Monitor. The supported DML attributes are:
   * <PRE>
   *   monitor [
   *     use SSF.Net.droptailQueueMonitor
   *     probe_interval  %F   # queue length averaging time interval, seconds (double)
   *     debug           %S   # print verbose diagnostics, true/false
   *   ]
   * </PRE>
   */
  public void config(NIC ownerNIC, Configuration cfg) throws configException {
    nic = ownerNIC;
    localHost = (Host)nic.inGraph();
    invfreq = 1.0/Net.frequency;

    // format time output
    fm.setMinimumFractionDigits(3);
    fm.setMaximumFractionDigits(3);
   
    // interval between record transmission, in seconds
    String str = (String)cfg.findSingle("probe_interval");
    if (str == null)
      throw new configException("\n"+nic.interfaceIdentifier()
                                +" DML attribute monitor.probe_interval must be specified");
    else {
      //tickinterval = Net.seconds((new Double(str)).doubleValue());
      intervalLength = (new Double(str)).doubleValue();
      tickinterval = Net.seconds(intervalLength);
    }

    str = (String)cfg.findSingle("debug");
    if (str != null)
      DEBUG = Boolean.valueOf(str).booleanValue();
  }

  /** called by SSF.Net.NIC.init()
   */
  public void init() {

    if (DEBUG)
      System.out.println(nic.interfaceIdentifier()
	                       + " droptailQueueMonitor_1: probe interval "
                         +tickinterval+" ticks");

    queue = (droptailQueue)nic.queueManager;  // add class verification
    bitrate = nic.BIT_RATE;

    try {
      P = (ProbeSession)(nic.inGraph().SessionForName("probe"));
      S = P.getRecorder();
      hostcode = P.getHostCode(nic.ID);
      typeCode = S.getRecordTypeCode("SSF.Net.QueueRecord_1");

      queue.setMonitor(this);

      packetQueueLog = new LinkedList();
      qResidual = 0;

      SSF.OS.Timer recordTimer = new SSF.OS.Timer(nic.inGraph(), tickinterval) {
        public void callback(){
          sendRecord();
          set(tickinterval);
        }
      };

      recordTimer.set();

    } catch (ProtocolException pex) {
      System.err.println(nic.interfaceIdentifier()+ " Can't configure droptailQueueMonitor_1: "
			 + "no ProbeSession registered in host");
      System.err.println(pex.getMessage());
      System.exit(1);
    }

    // write a single probe interval record for this NIC to the stream
    int probIntTypeCode = S.getRecordTypeCode("SSF.Net.QueueProbeIntRecord");
    double time = invfreq*localHost.now();
    byte[] intRecord = new byte[8];
    //    BytesUtil.doubleToBytes(intervalLength, intRecord, 0);
    /** FIXME: bytesToDouble missing in bytes util */
    BytesUtil.floatToBytes((float) intervalLength, intRecord, 0);
 
    S.send(probIntTypeCode, hostcode, time, intRecord, 0, intRecord.length);
  }

  private void sendRecord() {
    double time = invfreq*localHost.now();
    float qbytes = avgQueueLength();

    byte[] qRecord = new byte[12];
    int startIndex = 0;

    if (qbytes > 0) {	

      // check flaky drops
      if (nic.COUNT_FLAKY_DROPS) {
	drops += nic.flakyDrops - prevFlakyDrops; // add the flaky drops to 
				                  // the buffer drops
	prevFlakyDrops = nic.flakyDrops;
      }

      // output
      startIndex = BytesUtil.floatToBytes(qbytes, qRecord, startIndex);
      startIndex = BytesUtil.intToBytes(pkts, qRecord, startIndex);
      startIndex = BytesUtil.intToBytes(drops, qRecord, startIndex);

      if (DEBUG)
        System.out.println("QMon1: " + fm.format(time)
                           + " " + nic.interfaceIdentifier()
                           + " pkts " + pkts
                           + " drops " + drops
                           + " qlen " + qbytes);
      S.send(typeCode, hostcode, time, qRecord, 0, qRecord.length);
      pkts = 0;
      drops = 0;
    }
  }

  /** Log a packet record as a packet arrives to the queue.
   */
  public void receive(ProtocolMessage msg) {
    // PacketQueueRecord rec = new PacketQueueRecord(invfreq*localHost.now(),
	                                 // msg.bytecount()); //, queue.pktDropped);

    // packetQueueLog.add(rec); // append record to end of list
    pkts++;
    if(queue.pktDropped) {
      drops++;
    } else {
      PacketQueueRecord rec =
        new PacketQueueRecord(invfreq*localHost.now(), msg.bytecount()); //, queue.pktDropped);
      packetQueueLog.add(rec); // append record to end of list
    }
  }

  /** integral of queue_length(t) in the measurement interval
   *  divided by the length of the measurement interval.
   */
  public float avgQueueLength() {

    double endTime = localHost.now()*invfreq;
    double startTime = endTime - intervalLength;
    double area = 0;
    double lastTime = startTime;
    double nextTime, dT, resLen;
    PacketQueueRecord nextRecord;

    while(packetQueueLog.size() > 0) {
      nextRecord = (PacketQueueRecord)packetQueueLog.removeFirst();
      nextTime = nextRecord.arrivalTime;

      dT = Math.min(nextTime - lastTime, qlen/bitrate);
      resLen = Math.max(qlen - bitrate*dT, 0);  // residual queue length at nextTime-
      area = area + dT*resLen + 0.5*dT*(qlen - resLen);

      qlen = resLen + 8*nextRecord.size;        // queue length (bits) at nextTime+
      lastTime = nextTime;
    }

    // from last packet arrival to endTime
    dT = Math.min(endTime - lastTime, qlen/bitrate);
    resLen = Math.max(qlen - bitrate*dT, 0);   // residual queue length at endTime-
    area = area + dT*resLen + 0.5*dT*(qlen - resLen);
    qlen = resLen;

    return (float)(0.125*area/(endTime - startTime));  // avg. queue in bytes
  }
}


